package com.atyun.gaojin.notify.deduplicator.impl;


import com.atyun.gaojin.notify.deduplicator.AlterDeduplicator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.DigestUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 告警去重，此处未添加bean注解，可再config配置类为其生成bean对象
 * 基于Jaccard
 * @Author: 云水瑶
 * @Date: 2025/5/24
*/
@Slf4j
public class RedisAlertDeduplicator implements AlterDeduplicator {

    private final RedisTemplate<String, Object> redisTemplate;
    private final long deduplicationWindow;
    private final String HASH_KEY = "alter-hash-key-";
    private final String CONTENT_KEY = "alter-content";
    private final String TIMESTAMP = "timestamp";
    private final double MAX_VALUE = 0.9;

    public RedisAlertDeduplicator(RedisTemplate<String, Object> redisTemplate, long deduplicationWindow) {
        this.redisTemplate = redisTemplate;
        this.deduplicationWindow = deduplicationWindow;
    }

    /**
     * 检查告警是否应该去重
     * 待优化
     * 需知: redis的hash结构是一个复合结构
     * 第一个hash中，key为线程池所属的告警集合，key-value分别为md5content-content_value
     * 第二个hash中，key为md5content，key-value分别为content-timestamp
     * 第一个hash存储该线程池对应的所有告警，第二个为每个告警分别对应的内容和时间戳
     * @Author: 云水瑶
     * @Date: 2025/5/24
    */
    @Override
    public boolean shouldAlterDeduplication(String alterKey, String alterContent, String threadPoolName) {
        //需要根据这个key来判断是否存在于redis中从而判断是否去重
//        ValueOperations<String, Object> ops = redisTemplate.opsForValue();
//        Boolean flag = ops.setIfAbsent(alterKey, "1", deduplicationWindow, TimeUnit.SECONDS);
        HashOperations<String, String, String> ops = redisTemplate.opsForHash();
        String contentValue = ops.get(alterKey, CONTENT_KEY);
        if(alterContent.equals(contentValue)) {
            log.info("出现完全相同的内容告警，不需要再发告警: {}", alterKey);
            return false;//出现完全相同的内容告警，不需要再发告警，返回false
        }

        //构建线程池hashKey，取出线程池对应的所有告警，进行jaccard匹配
        String hashKey = HASH_KEY + threadPoolName;
        //key为alterKey，value为content
        Map<String, String> allAlter = ops.entries(hashKey);
        for(Map.Entry<String, String> entry : allAlter.entrySet()) {
            String value = entry.getValue();
            //进行相似度匹配
            double num = jaccard(alterContent, value);
            log.info(alterKey + "与" + entry.getKey() + "匹配的相似度为：" + num);
            if(num >= MAX_VALUE) {
                //相似度超过max值90%，内容非常相似，需要去重
                return false;
            }
        }
        //经过了内容与相似度的匹配，代码执行到这里说明可以发布告警，需要存储
        Map<String, String> alterDataMap = new ConcurrentHashMap<>();
        //存储第二个hash的content与timestamp
        alterDataMap.put(CONTENT_KEY, alterContent);
        alterDataMap.put(TIMESTAMP, String.valueOf(System.currentTimeMillis()));
        ops.putAll(alterKey, alterDataMap);
        //为该hash（单个告警）设置过期时间
        redisTemplate.expire(alterKey, deduplicationWindow, TimeUnit.SECONDS);

        //给该告警存一份hash到其对所属的线程池hash结构中（存入第一个hash中）
        ops.put(hashKey, alterKey, alterContent);
        return true;
    }

    /**
     * 相似度计算，jaccard：交集/并集
     * @Author: 云水瑶
     * @Date: 2025/5/25
    */
    private double jaccard(String text1, String text2) {
        //先分词，存入set中
        Set<String> set1 = tokenization(text1);
        Set<String> set2 = tokenization(text2);
        //计算相似度
        Set<String> intersectSet = new HashSet<>(set1);
        intersectSet.retainAll(set2); //计算交集
        Set<String> unionSet = new HashSet<>(set1);
        unionSet.addAll(set2);
        return unionSet.isEmpty() ? 0 : (double) intersectSet.size() / unionSet.size();
    }

    /**
     * 分词方法，基于正则表达式
     * @Author: 云水瑶
     * @Date: 2025/5/25
    */
    private Set<String> tokenization(String text) {
        if(!StringUtils.hasText(text)) {
            return Collections.emptySet();
        }
        String[] strArr = text.toLowerCase().split("\\W+");
        return Arrays.stream(strArr).filter(word -> !word.isEmpty()).collect(Collectors.toSet());
    }

    /**
     * 定时任务：定时扫描每个线程池缓存的告警，并清除过期的告警
     * 每天凌晨3.30执行
     * @Author: 云水瑶
     * @Date: 2025/5/25
    */
    @Scheduled(cron = "0 30 3 * * ?")
    public void taskDeleteOverAlter() {
        List<String> list = new ArrayList<>();
        Set<String> allPoolHashKeys = redisTemplate.keys(HASH_KEY + "*");
        if(Objects.isNull(allPoolHashKeys) || allPoolHashKeys.isEmpty()) return;

        //删除超过5分钟的告警
        long time = System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(5 * 60);
        HashOperations<String, String, String> ops = redisTemplate.opsForHash();
        for(String poolKey : allPoolHashKeys) {
            //获取该线程池下所有的告警md5
            Map<String, String> allAlterMD5 = ops.entries(poolKey);
            for(Map.Entry<String, String> entry : allAlterMD5.entrySet()) {
                String alterKey = entry.getKey();
                String timestamp = ops.get(alterKey, TIMESTAMP);
                if(StringUtils.isEmpty(timestamp) || Long.parseLong(timestamp) <= time) {
                    //过期了，先整合在一起，然后再清除
                    list.add(alterKey);
                }
            }
            if(!list.isEmpty()) {
                log.info("需要删除的过期告警md5码: {}", list);
                ops.delete(poolKey, list.toArray());
                list = new ArrayList<>();
            }
        }
    }

    /**
     * 生产alterKey
     * @Author: 云水瑶
     * @Date: 2025/5/24
    */
    @Override
    public String getAlterKey(String threadPoolName, String alterContent) {
        String primalKey = threadPoolName + "-|-" + alterContent;
        String alterKeyMD5 = DigestUtils.md5DigestAsHex(primalKey.getBytes());
        return "alter:deduplication:" + alterKeyMD5;
    }
}
